package drds.binlog.parse.inbound.mysql;

import drds.binlog.*;
import drds.binlog.event.QueryLogEvent;
import drds.binlog.parse.driver.packets.GtidSet;
import drds.binlog.parse.exception.BinlogParseException;
import drds.binlog.parse.exception.ServerIdNotMatchException;
import drds.binlog.parse.inbound.Connection;
import drds.binlog.parse.inbound.IMultiStageCoprocessor;
import drds.binlog.parse.inbound.SinkFunction;
import drds.binlog.parse.inbound.mysql.local.BinLogFileQueue;
import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.List;

/**
 * local bin log connection (not real connection)
 */
public class LocalBinLogConnection implements Connection
{

    private static final Logger logger = LoggerFactory.getLogger(LocalBinLogConnection.class);
    private BinLogFileQueue binLogFileQueue = null;
    private boolean needWait;
    private String directory;
    private int bufferSize = 16 * 1024;
    private boolean running = false;
    private long serverId;
    private FileParserListener fileParserListener;

    public LocalBinLogConnection()
    {
    }

    public LocalBinLogConnection(String directory, boolean needWait)
    {
        this.needWait = needWait;
        this.directory = directory;
    }

    @Override
    public void connect() throws IOException
    {
        if (this.binLogFileQueue == null)
        {
            this.binLogFileQueue = new BinLogFileQueue(this.directory);
        }
        this.running = true;
    }

    @Override
    public void reconnect() throws IOException
    {
        disconnect();
        connect();
    }

    @Override
    public void disconnect() throws IOException
    {
        this.running = false;
        if (this.binLogFileQueue != null)
        {
            this.binLogFileQueue.destory();
        }
        this.binLogFileQueue = null;
        this.running = false;
    }

    public boolean isConnected()
    {
        return running;
    }

    public void seek(String binlogFileName, Long binlogPosition, SinkFunction sinkFunction) throws IOException
    {
    }

    public void dump(String binlogFileName, Long binlogPosition, SinkFunction sinkFunction) throws IOException
    {
        File current = new File(directory, binlogFileName);

        FileLogFetcher fileLogFetcher = new FileLogFetcher(bufferSize);
        LogDecoder logDecoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
        LogContext logContext = new LogContext();
        try
        {
            fileLogFetcher.open(current, binlogPosition);
            logContext.setLogPosition(new LogPosition(binlogFileName, binlogPosition));
            while (running)
            {
                boolean needContinue = true;
                LogEvent logEvent = null;
                while (fileLogFetcher.fetch())
                {
                    logEvent = logDecoder.decode(fileLogFetcher, logContext);
                    if (logEvent == null)
                    {
                        continue;
                    }
                    if (serverId != 0 && logEvent.getServerId() != serverId)
                    {
                        throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !");
                    }

                    if (!sinkFunction.sink(logEvent))
                    {
                        needContinue = false;
                        break;
                    }
                }

                fileLogFetcher.close(); // 关闭上一个文件
                parserFinish(current.getName());
                if (needContinue)
                {// 读取下一个

                    File nextFile;
                    if (needWait)
                    {
                        nextFile = binLogFileQueue.waitForNextFile(current);
                    } else
                    {
                        nextFile = binLogFileQueue.getNextFile(current);
                    }

                    if (nextFile == null)
                    {
                        break;
                    }

                    current = nextFile;
                    fileLogFetcher.open(current);
                    logContext.setLogPosition(new LogPosition(nextFile.getName()));
                } else
                {
                    break;// 跳出
                }
            }
        } catch (InterruptedException e)
        {
            logger.warn("LocalBinLogConnection dump interrupted");
        } finally
        {
            if (fileLogFetcher != null)
            {
                fileLogFetcher.close();
            }
        }
    }

    public void dump(long timestampMills, SinkFunction sinkFunction) throws IOException
    {
        List<File> currentBinlogs = binLogFileQueue.currentBinlogs();
        File current = currentBinlogs.get(currentBinlogs.size() - 1);
        long timestampSeconds = timestampMills / 1000;

        String binlogFilename = null;
        long binlogFileOffset = 0;

        FileLogFetcher fileLogFetcher = new FileLogFetcher(bufferSize);
        LogDecoder logDecoder = new LogDecoder();
        logDecoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);
        logDecoder.handle(LogEvent.QUERY_EVENT);
        logDecoder.handle(LogEvent.XID_EVENT);
        LogContext logContext = new LogContext();
        try
        {
            fileLogFetcher.open(current);
            logContext.setLogPosition(new LogPosition(current.getName()));
            while (running)
            {
                boolean needContinue = true;
                String lastXidLogFilename = current.getName();
                long lastXidLogFileOffset = 0;

                binlogFilename = lastXidLogFilename;
                binlogFileOffset = lastXidLogFileOffset;
                while (fileLogFetcher.fetch())
                {
                    LogEvent event = logDecoder.decode(fileLogFetcher, logContext);
                    if (event != null)
                    {
                        if (serverId != 0 && event.getServerId() != serverId)
                        {
                            throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !");
                        }

                        if (event.getWhen() > timestampSeconds)
                        {
                            break;
                        }

                        needContinue = false;
                        if (LogEvent.QUERY_EVENT == event.getLogHeader().getType())
                        {
                            if (StringUtils.endsWithIgnoreCase(((QueryLogEvent) event).getQuery(), "BEGIN"))
                            {
                                binlogFilename = lastXidLogFilename;
                                binlogFileOffset = lastXidLogFileOffset;
                            } else if (StringUtils.endsWithIgnoreCase(((QueryLogEvent) event).getQuery(), "COMMIT"))
                            {
                                lastXidLogFilename = current.getName();
                                lastXidLogFileOffset = event.getLogPosition();
                            }
                        } else if (LogEvent.XID_EVENT == event.getLogHeader().getType())
                        {
                            lastXidLogFilename = current.getName();
                            lastXidLogFileOffset = event.getLogPosition();
                        } else if (LogEvent.FORMAT_DESCRIPTION_EVENT == event.getLogHeader().getType())
                        {
                            lastXidLogFilename = current.getName();
                            lastXidLogFileOffset = event.getLogPosition();
                        }
                    }
                }

                if (needContinue)
                {// 读取下一个
                    fileLogFetcher.close(); // 关闭上一个文件

                    File nextFile = binLogFileQueue.getBefore(current);
                    if (nextFile == null)
                    {
                        break;
                    }

                    current = nextFile;
                    fileLogFetcher.open(current);
                    logContext.setLogPosition(new LogPosition(current.getName()));
                } else
                {
                    break;// 跳出
                }
            }
        } finally
        {
            if (fileLogFetcher != null)
            {
                fileLogFetcher.close();
            }
        }

        dump(binlogFilename, binlogFileOffset, sinkFunction);
    }

    @Override
    public void dump(GtidSet gtidSet, SinkFunction sinkFunction) throws IOException
    {
        throw new NotImplementedException();
    }

    @Override
    public void dump(String binlogFileName, Long binlogPosition, IMultiStageCoprocessor IMultiStageCoprocessor) throws IOException
    {
        File binlogFile = new File(directory, binlogFileName);
        if (!binlogFile.exists())
        {
            throw new BinlogParseException("binlog:" + binlogFileName + " is not found");
        }

        FileLogFetcher fileLogFetcher = new FileLogFetcher(bufferSize);
        LogDecoder logDecoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
        LogContext logContext = new LogContext();
        try
        {
            fileLogFetcher.open(binlogFile, binlogPosition);
            logContext.setLogPosition(new LogPosition(binlogFileName, binlogPosition));
            while (running)
            {
                boolean needContinue = true;
                LogEvent logEvent = null;
                while (fileLogFetcher.fetch())
                {
                    logEvent = logDecoder.decode(fileLogFetcher, logContext);
                    if (logEvent == null)
                    {
                        continue;
                    }
                    if (serverId != 0 && logEvent.getServerId() != serverId)
                    {
                        throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !");
                    }

                    if (!IMultiStageCoprocessor.publish(logEvent))
                    {
                        needContinue = false;
                        break;
                    }
                }

                fileLogFetcher.close(); // 关闭上一个文件
                parserFinish(binlogFileName);
                if (needContinue)
                {// 读取下一个
                    File nextFile;
                    if (needWait)
                    {
                        nextFile = binLogFileQueue.waitForNextFile(binlogFile);
                    } else
                    {
                        nextFile = binLogFileQueue.getNextFile(binlogFile);
                    }

                    if (nextFile == null)
                    {
                        break;
                    }

                    binlogFile = nextFile;
                    fileLogFetcher.open(binlogFile);
                    binlogFileName = nextFile.getName();
                } else
                {
                    break;// 跳出
                }
            }
        } catch (InterruptedException e)
        {
            logger.warn("LocalBinLogConnection dump interrupted");
        } finally
        {
            if (fileLogFetcher != null)
            {
                fileLogFetcher.close();
            }
        }
    }

    private void parserFinish(String fileName)
    {
        if (fileParserListener != null)
        {
            fileParserListener.onFinish(fileName);
        }
    }

    @Override
    public void dump(long timestampMills, IMultiStageCoprocessor IMultiStageCoprocessor) throws IOException
    {
        List<File> currentBinlogs = binLogFileQueue.currentBinlogs();
        File current = currentBinlogs.get(currentBinlogs.size() - 1);
        long timestampSeconds = timestampMills / 1000;

        String binlogFilename = null;
        long binlogFileOffset = 0;

        FileLogFetcher fileLogFetcher = new FileLogFetcher(bufferSize);
        LogDecoder decoder = new LogDecoder();
        decoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);
        decoder.handle(LogEvent.QUERY_EVENT);
        decoder.handle(LogEvent.XID_EVENT);
        LogContext context = new LogContext();
        try
        {
            fileLogFetcher.open(current);
            context.setLogPosition(new LogPosition(current.getName()));
            while (running)
            {
                boolean needContinue = true;
                String lastXidLogFilename = current.getName();
                long lastXidLogFileOffset = 0;

                binlogFilename = lastXidLogFilename;
                binlogFileOffset = lastXidLogFileOffset;
                while (fileLogFetcher.fetch())
                {
                    LogEvent event = decoder.decode(fileLogFetcher, context);
                    if (event != null)
                    {
                        if (serverId != 0 && event.getServerId() != serverId)
                        {
                            throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !");
                        }

                        if (event.getWhen() > timestampSeconds)
                        {
                            break;
                        }

                        needContinue = false;
                        if (LogEvent.QUERY_EVENT == event.getLogHeader().getType())
                        {
                            if (StringUtils.endsWithIgnoreCase(((QueryLogEvent) event).getQuery(), "BEGIN"))
                            {
                                binlogFilename = lastXidLogFilename;
                                binlogFileOffset = lastXidLogFileOffset;
                            } else if (StringUtils.endsWithIgnoreCase(((QueryLogEvent) event).getQuery(), "COMMIT"))
                            {
                                lastXidLogFilename = current.getName();
                                lastXidLogFileOffset = event.getLogPosition();
                            }
                        } else if (LogEvent.XID_EVENT == event.getLogHeader().getType())
                        {
                            lastXidLogFilename = current.getName();
                            lastXidLogFileOffset = event.getLogPosition();
                        } else if (LogEvent.FORMAT_DESCRIPTION_EVENT == event.getLogHeader().getType())
                        {
                            lastXidLogFilename = current.getName();
                            lastXidLogFileOffset = event.getLogPosition();
                        }
                    }
                }

                if (needContinue)
                {// 读取下一个
                    fileLogFetcher.close(); // 关闭上一个文件

                    File nextFile = binLogFileQueue.getBefore(current);
                    if (nextFile == null)
                    {
                        break;
                    }

                    current = nextFile;
                    fileLogFetcher.open(current);
                    context.setLogPosition(new LogPosition(current.getName()));
                } else
                {
                    break;// 跳出
                }
            }
        } finally
        {
            if (fileLogFetcher != null)
            {
                fileLogFetcher.close();
            }
        }

        dump(binlogFilename, binlogFileOffset, IMultiStageCoprocessor);
    }

    @Override
    public void dump(GtidSet gtidSet, IMultiStageCoprocessor IMultiStageCoprocessor) throws IOException
    {
        throw new NotImplementedException();
    }

    public Connection fork()
    {
        LocalBinLogConnection connection = new LocalBinLogConnection();

        connection.setBufferSize(this.bufferSize);
        connection.setDirectory(this.directory);
        connection.setNeedWait(this.needWait);
        return connection;
    }

    @Override
    public long queryServerId()
    {
        return 0;
    }

    public boolean isNeedWait()
    {
        return needWait;
    }

    public void setNeedWait(boolean needWait)
    {
        this.needWait = needWait;
    }

    public String getDirectory()
    {
        return directory;
    }

    public void setDirectory(String directory)
    {
        this.directory = directory;
    }

    public int getBufferSize()
    {
        return bufferSize;
    }

    public void setBufferSize(int bufferSize)
    {
        this.bufferSize = bufferSize;
    }

    public long getServerId()
    {
        return serverId;
    }

    public void setServerId(long serverId)
    {
        this.serverId = serverId;
    }

    public void setFileParserListener(FileParserListener fileParserListener)
    {
        this.fileParserListener = fileParserListener;
    }

    public interface FileParserListener
    {

        void onFinish(String fileName);
    }

}
